package yunjiao.javatutorials.guava.concurrent;

import com.google.common.util.concurrent.AbstractExecutionThreadService;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 消息消费者服务
 *
 * @author yangyunjiao
 */
public class MessageConsumerService extends AbstractExecutionThreadService {
    // 消息队列（模拟）
    private final BlockingQueue<String> messageQueue = new LinkedBlockingQueue<>();
    // 服务名称（自定义）
    @Override
    protected String serviceName() {
        return "MessageConsumerService";
    }

    // 启动前初始化：模拟连接消息队列
    @Override
    protected void startUp() throws Exception {
        System.out.printf("[%s] 启动初始化：连接消息队列成功%n", serviceName());
        // 模拟添加测试消息
        messageQueue.offer("测试消息1");
        messageQueue.offer("测试消息2");
    }

    // 核心业务逻辑：持续消费消息
    @Override
    protected void run() throws Exception {
        System.out.printf("[%s] 开始消费消息（状态：%s）%n", serviceName(), state());
        while (isRunning()) { // 以服务状态作为退出条件
            try {
                // 阻塞获取消息（1秒超时，避免无限阻塞）
                String message = messageQueue.poll(1, TimeUnit.SECONDS);
                if (message != null) {
                    System.out.printf("[%s] 消费消息：%s%n", serviceName(), message);
                }
                // 响应中断（stopAsync()会中断当前线程）
                if (Thread.interrupted()) {
                    System.out.printf("[%s] 检测到中断信号，准备停止%n", serviceName());
                    break;
                }
            } catch (InterruptedException e) {
                System.out.printf("[%s] 线程被中断，退出消费%n", serviceName());
                Thread.currentThread().interrupt(); // 恢复中断状态
                break;
            }
        }
    }

    // 停止后清理：模拟关闭消息队列连接
    @Override
    protected void shutDown() throws Exception {
        System.out.printf("[%s] 停止清理：关闭消息队列连接成功%n", serviceName());
    }

    // 测试代码
    public static void main(String[] args) throws InterruptedException, TimeoutException {
        MessageConsumerService service = new MessageConsumerService();

        // 1. 启动服务（异步）
        service.startAsync();
        service.awaitRunning(); // 阻塞等待服务启动完成
        System.out.printf("[主线程] 服务已启动（状态：%s）%n", service.state());

        // 2. 运行3秒后停止服务
        TimeUnit.SECONDS.sleep(3);
        service.stopAsync(); // 异步停止
        System.out.printf("[主线程] 已触发服务停止（状态：%s）%n", service.state());

        // 3. 等待服务终止（最多等待2秒）
        service.awaitTerminated(2, TimeUnit.SECONDS);
        System.out.printf("[主线程] 服务正常终止（状态：%s）%n", service.state());

    }
}
